package peerToPeerFacebook.proj;

import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import peerToPeerFacebook.proj.DataManager.TransactionStatus;
import peerToPeerFacebook.proj.Log.LogRecord;
import peerToPeerFacebook.src.edu.washington.cs.cse490h.lib.Callback;

/** This class provides implementation of paxos state machine and paxos learner concept. */
public class PaxosStateMachine
{
    /** Name of data manager transaction */
    protected static final String DataManagerTransactionID = "psm-transaction";

    /** Class that allows to store command that this node still needs to execute. Also stores CallPending instance
     * that client is querying to figure out the result.
     */
    protected class PendingCommand
    {
        /** Command that needs to be executed */
        public PaxosCommand command = null;
        
        /** CallPending that should be used to return result to client */
        public CallPendingImpl callback = null;
        
        /** Simple constructor */
        public PendingCommand(PaxosCommand cmd, CallPendingImpl cp)
        {
            this.command = cmd;
            this.callback = cp;
        }
    }
    
    /** Defines state of the PSM. This basically makes sure that only one prepare call can be outstanding. The problem
     * is that we have multiple prepare statements issued, we will not be able to figure out if out, which our command
     * was accepted. */
    protected enum PrepareState {
        /** We are waiting result from some prepare call. We cannot make any more prepares at this time */
        Waiting,
        
        /** We finished all prepare calls tha we had and we are ready to do more of them */
        Ready
    }
    
    /** Log records that this class will write to the log */
    protected enum LogRecords {
        /** Param: newCommnadID, clientID, sequenceID . Record may not be complete */
        LastCommandID
    }
    
    /** Command ID that was last executed. This one needs to always be serialized to the disk. */
    protected int lastCommandID = 0;
    
    /** A list of commands that are accepted by network but not applied on this node. This list is not empty iff 
     * we know that there are some commands that we missed.
     * Key -> Command ID, value -> Command for this round.
     * 
     * We do not store this list to the hard drive. Acceptors does that for us and we can always re-learn values from
     * them.
     */
    protected Map<Integer, PaxosCommand> commandsToApply = new HashMap<Integer, PaxosCommand>();
    
    /** list of commands that we received for client that we have not executed yet. We execute commands in list in the
     * same order they were added.
     * 
     *  We do not persist this list in the hard disk. If we crash, client will receive call results of AtMostOnce and
     *  it is up to it to decide what to do.
     */
    protected List<PendingCommand> pendingCommands = new LinkedList<PendingCommand>();
    
    /** Node class used for IO operations */
    protected FBNode node = null;
    
    /** Defines if PSM can issue new prepare call */
    protected PrepareState prepareState = PrepareState.Ready;
    
    /** Log used to store command IDs */
    protected Log log = null;
    
    /** Command that we are currently trying to propse */
    protected PaxosCommand activeProposal = null;
    
    /** Sequance ID that was last used for given client. IN case older sequance ID is seen it will de dropped */
    protected Map<Integer, Integer> clientSequanceID = new HashMap<Integer, Integer>();
    
    public PaxosStateMachine(FBNode node) throws IOException
    {
        this.node = node;
        this.lastCommandID = 0;
        
        log = new Log(new NodeWrapper(node), "psm-log");
        handleLastOperation();
    }

    /** Handles last operation that this node tried to execute and updated lastCommandID. This method will read log file
     * and handles possibly unfinished last command. However, if command failed before Pre-Commit state than, it will be
     * aborted. In such case this ID will not be restored and PSM will need to discover that it is missing this command. 
     * 
     */
    protected void handleLastOperation() throws IOException
    {
        LogRecord lastRecord = null;
        // read the log and find out the last ID
        for (LogRecord record: log.read())
        {
            assert lastRecord == null || (lastRecord != null && lastRecord.completed):
                "Only last record can be incomplete";
            
            String[] params = record.param.split(" ");
            int id = Integer.parseInt(params[0]);
            assert lastCommandID < id: "Command IDs should be in increasing order";
            lastCommandID = id;
            
            // if command was applied sucesfully update client IDs. Otherwise we will need to learn this command again.
            if (record.completed)
                clientSequanceID.put(Integer.parseInt(params[1]), Integer.parseInt(params[2]));
            
            lastRecord = record;
        }
        
        // Instantiate DM. If last record is complete that it is possible that there is some stale data, that DM
        // needs to clean up. It will do that automatically by aborting transaction. Which is what we want if we failed
        // before we noted about applying our transaction ID. We deal with other situations a bit later.
        // However, if DM log does not exists that means that we have committed it already and were just unable to finish
        // last record
        DataManager dm = null;
        if (DataManager.logFileExists(new NodeWrapper(node), DataManagerTransactionID))
        {
            dm = createDataManager();
        
            // We may need to abort this transaction. If we moved it to into PreCommit state, but failed to store ID
            // we do not know what ID is associated with this command. We will allow regular PSM gap detection to deal
            // with this.
            if ((lastRecord == null || lastRecord.completed) && dm.getSatus() == TransactionStatus.PreCommit)
               dm.abort();
            
            // if transaction was (possible auto) aborted, we are done. Regular PSM gap detection will detect that we are
            // missing this ID and will discover it otherwise.
            if (dm.getSatus() == TransactionStatus.Aborted)
            {
                assert lastRecord == null || lastRecord.completed: "Last record should be completed as it signals " +
                		"old ID, not current one";
                return;
            }
        }
        
        // DM can still be in PreCommit mode if we managed to write out the record for this transaction. Otherwise DM is
        // null and we are done
        assert dm == null || dm.getSatus() == TransactionStatus.PreCommit: "We either committed (dm == null) or in " +
        		"PreCommit state. All other states should have been handled";
        assert dm == null || !lastRecord.completed: "If we are in PreCommit state (dm != null) lastRecord should not " +
        		"be completed";
        
        // Finish committing action if we need to.
        if (dm != null && dm.getSatus() == TransactionStatus.PreCommit)
            dm.commit();
        
        // now we need to do house keeping. At this point either we decided that transaction is committed or we done that
        // so finish record
        if (lastRecord != null && !lastRecord.completed)
            log.writeEnd(LogRecords.LastCommandID.toString());
    }
    
    /** This method is called by paxos proposer upon new value for some particular command is learned
     * 
     * @param commandID Command for which value is learn
     * @param value Value that has been discovered
     * @param ourProposal True if this is command that this node propose before. This means that we also need to ping
     *  our and tell it about outcome.
     */
    public void learn(int commandID, Serializable value, boolean ourProposal)
    {
        System.out.println("PSM: New command. ID: " + commandID + ", ours: " + ourProposal + ", value: " + value);
        assert !((PaxosCommand)value).isNOP(): "We should never be told about NOP command. It should only be used to " +
        		"learn already existing commands.";
        
        // It is possible that we already applied this command or we know about it. If so we just skip it.
        if (commandID <= lastCommandID)
            return;
        
        // Is this command that we have stashed to apply later?
        if (commandsToApply.containsKey(commandID))
        {
            assert commandsToApply.get(commandID).equals(value): "This node thinks that this command should be different";
            return;
        }
        
        // Are we expecting our command to be accepted?
        if (activeProposal != null)
            // Maybe this is our command?
            if (ourProposal)
            {
                // From now on, other prepare statements are OK
                prepareState = PrepareState.Ready;
                PendingCommand pc = pendingCommands.remove(0);
    
                assert commandID == lastCommandID + 1: "We always should be able to execute our command immediately";
                assert pc.command.equals(value): "Our command should always be on the top of the list " +
                		"of pending commands";
                assert pc.command.equals(activeProposal): "We should have active proposal";
                activeProposal = null;
                
                // Try to apply the command and see what happens. If we get some funky results such as exception, we
                // just send them to client.
                try
                {
                    Serializable result = applyCommand((PaxosCommand)value);
                    pc.callback.setResult(result);
                } catch (Throwable e)
                {
                    pc.callback.setException(e);
                }
    
            }
            else
            {
                // well it is not our proposal... executePendingCommands will retry for us. We just need to clean the
                // state. We do not remove commands from the list though.
                activeProposal = null;
                prepareState = PrepareState.Ready;
                commandsToApply.put(commandID, (PaxosCommand)value);
            }  
        else
            commandsToApply.put(commandID, (PaxosCommand)value);
        
        // See if we can process some other commands. The problem is that we can't execute commands now. If this our
        // proposal, that means that we sending this call to all other paxos servers - thus using RPC channel.
        executePendingCommands();
    }
    
    /** Executes user provided command. As command can be executed only after the paxos reaches consensus this function
     * always returns call pending.
     * 
     * @param cmd Command to execute.
     * @return Call pending object that should be used to query for result.
     */
    public CallPending execute(PaxosCommand cmd)
    {
        PendingCommand pc = new PendingCommand(cmd, new CallPendingImpl());
        pendingCommands.add(pc);
        
        // See if we can execute it immediately.
        if (commandsToApply.size() == 0 && prepareState == PrepareState.Ready)
            proposeCommand(pc.command, lastCommandID + 1);
        
        return pc.callback;
    }
    
    /** This method checks what was the last command we had and what commands are still pending. Than it it will try
     * to learn missing ones and execute them.
     * 
     * Public as this one is being invoked from callback
     */
    public void executePendingCommands()
    {
        // Apply as much commands as we can
        while(commandsToApply.containsKey(lastCommandID + 1))
            try
            {
                applyCommand(commandsToApply.remove(lastCommandID + 1));
            } catch (Throwable e)
            {
                // we ignore all exceptions. We assume that client can deal with them and it was notified about them
                // All exceptions here are the ones caused by client.
            }
        
        // From this point on, we try to prepare some of the commands. We will do that only if we were called in
        // correct state. We are called after each learn is done, so it we can just skip now if we are not ready and
        // finish work later.
        
        // Do we still have some gaps?
        if (commandsToApply.size() == 0)
        {
            // Nothing to apply, let's see if we have our commands to run
            if (pendingCommands.size() > 0 && prepareState == PrepareState.Ready)
                proposeCommand(pendingCommands.get(0).command, lastCommandID + 1);
            
            // Otherwise we are done
            return;
        }
        
        // ok we have some gaps, that we need to learn, let's start asking about them by suggesting NOP
        // Ideally we should never get back NOP.
        if (prepareState == PrepareState.Ready)
            proposeCommand(PaxosCommand.nop(), lastCommandID + 1);
    }
    
    /** Apples given command to local instance. This method is called after Paxos reaches consensus on command to
     * execute.
     * 
     * @param cmd Command to execute.
     * @return return of the command execution.
     */
    protected Serializable applyCommand(PaxosCommand cmd) throws Throwable
    {
        System.out.println("PSM: applying command. ID: " + (lastCommandID + 1) + ", command: " + cmd);
        // see if we have executed this command first
        if (clientSequanceID.containsKey(cmd.clientID) && cmd.sequenceID <= clientSequanceID.get(cmd.clientID))
        {
            // We do know the answer at this point, so we throw the exception and move on.
            System.out.println("PSM: This command was already executed. Ignoring");
            throw new RPCError("Command already executed");
        }
        
        // Construct DM. This is just temporary object that will be used only inside this method
        // thus we do not care too much about any global properties.
        DataManager dm = createDataManager();
        DynamicInvoker invoker = new DynamicInvoker(dm);
        
        // execute command
        Serializable result = null;
        Throwable ex = null;
        
        try
        {
           result = (Serializable) invoker.invoke(cmd.method, cmd.params);
        } catch (InvocationTargetException e) {
            // Even RuntimeException and Errors exceptions from target will generate this exception, so we specifically
            // filter them out and let them pass - this sort of simulates server crash.
            Throwable t = e.getCause();
            if (t instanceof RuntimeException)
                throw (RuntimeException)t;
            else if (t instanceof Error)
                throw (Error)t;
            
            ex = e;
        } catch (Throwable e1)
        {
            ex = e1;
        }

        // need to make sure that we complete our book-keeping if call crashes.
        dm.preCommit();

        // If the crash happened before this point - we either will not know that command existed of DM will
        // automatically abort it or we do that. However, before we start doing commit we need to have a trace that
        // command was started to be executed. This is required as after the commit DM log is deleted and on restart we
        // will not be able to tell if command was executed or not.
        lastCommandID++;
        log.writeStart(LogRecords.LastCommandID.toString(),
                Integer.toString(lastCommandID) + " " + cmd.clientID + " " + cmd.sequenceID);
        
        // now do commit and hope that we are good
        dm.commit();
        
        // IF we are here, that means that command completed successfully and log file is gone. Now we can finish our
        // book keeping and tell the client good news.
        log.writeEnd(LogRecords.LastCommandID.toString());       
        
        // Mark command as executed
        if (clientSequanceID.containsKey(cmd.clientID) && clientSequanceID.get(cmd.clientID) != cmd.sequenceID - 1)
            System.out.println("PSM: DBG: Client sent commands not in sequence.");
        clientSequanceID.put(cmd.clientID, cmd.sequenceID);

        // if we had an exception we need to re-throw it, to let RPC return it back to the caller
        if (ex != null)
            throw ex;
        
        return result;
    }

    /** Creates data manager that should be used for IO operations
     * 
     * @return Data manager to use for IO operations
     */
    protected DataManager createDataManager() throws IOException
    {
        return new DataManager(DataManagerTransactionID, new NodeWrapper(node), new LockManager());
    }

    /** Tries to propose given command as next command in paxos network.
     * 
     * @param cmd Command to propose.
     */
    protected void proposeCommand(PaxosCommand cmd, int commandID)
    {
        // Make sure that RPC clients are clear before executing command. if not, delay execution
        boolean clean = true;
        for (int i : FBNode.PaxosGroup)
        {
            RPCClient client = this.node.getRPCClient(i);
            clean = clean && !client.callPending();
        }
        
        if (!clean)
        {
            try
            {
                Method method = Callback.getMethod("executePendingCommands", this, new String[] { } );
                Callback cb = new Callback(method, this, new Object[] { });
                node.addTimeout(cb, 1);

            } catch (ClassNotFoundException e)
            {
                assert false: "Failed to set callback";
            } catch (NoSuchMethodException e)
            {
                assert false: "Failed to set callback";
            } catch (SecurityException e)
            {
                assert false: "Failed to set callback";
            }

            return;
        }
        
        System.out.println("PSM: Proposing command. ID: " + commandID + ", command: " + cmd );
        assert commandsToApply.size() == 0 || cmd.isNOP(): "We can only propose new commands when we learned all the gaps";
        assert prepareState == PrepareState.Ready: "New prepare calls can be issued only when all others completed";
        assert activeProposal == null: "We should not have active proposal";
        
        // Move to waiting state
        prepareState = PrepareState.Waiting;
        activeProposal = cmd;
        
        // Try to convince paxos proposer to push our message
        PaxosProposer proposer = new PaxosProposer(node, FBNode.PaxosGroup);
        proposer.propose(cmd, commandID);
        
        // At this point there is nothing for us to do. We just need to wait for Proposer call out learn method
        // at some point.
    }    
}
